package com.xujianlong.app.dwm;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;
import com.xujianlong.utils.MyKafkaUtil;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //TODO 2.读取kafka数据
        String sourceTopic = "dwd_page_log";
        String groupId = "user_jump_detail_app_210726";
        String sinkTopic = "dwm_user_jump_detail";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaSource(sourceTopic, groupId));
        // TODO 3.转换为json
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSONObject.parseObject(s);
            }
        });
        //TODO 4.加上watermark
        SingleOutputStreamOperator<JSONObject> jsonObjectWithWaterMarkDS = jsonObjectDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        return jsonObject.getLong("ts");
                    }
                })

        );
        //TODO 5.keyby
        KeyedStream<JSONObject, String> keyedStream = jsonObjectWithWaterMarkDS.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));

        //TODO 6.创建模板
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").
                where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject, Context<JSONObject> context) throws Exception {
                        return jsonObject.getJSONObject("page").getString("last_page_id") == null;
                    }
                })
                .next("next").where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject, Context<JSONObject> context) throws Exception {
                        return jsonObject.getJSONObject("page").getString("last_page_id") == null;
                    }
                }).within(Time.seconds(10));
        //TODO 7.作用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);
        //TODO 8.选择结果
        OutputTag<JSONObject> timeOut = new OutputTag<JSONObject>("timeOut"){};
        SingleOutputStreamOperator<JSONObject> selectedDS = patternStream.select(timeOut, new PatternTimeoutFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
                return map.get("start").get(0);
            }
        }, new PatternSelectFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                return map.get("start").get(0);
            }
        });
        selectedDS.print("select>>>>>");
        DataStream<JSONObject> sideOutput = selectedDS.getSideOutput(timeOut);
        sideOutput.print("timeout>>>>");
        //TODO 9.合并流
        DataStream<JSONObject> unionDS = selectedDS.union(sideOutput);
        //TODO 10.输出到kafka
        unionDS.map(jsonObject -> jsonObject.toJSONString()).addSink(MyKafkaUtil.getKafkaSink(sinkTopic));
        //TODO 11.启动任务

        env.execute();
    }
}
